package com.seaboxdata.threadHandler;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ser.std.StringSerializer;
import com.seaboxdata.entity.BlockMileage;
import com.seaboxdata.kafka.BaseStringKafkaConsumer;
import com.seaboxdata.service.BlockMileageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;

@Component
@Slf4j
public class YtcHandler  implements DisposableBean, Runnable{

    private  KafkaConsumer<String, String> consumer;
    private  String topic;


//    @Value("${real.trans.ytcTopic}")
//    private String topic;

    @Value("${interface.isTest}")
    private String isTest;

    private boolean flag;

    @Autowired
    public void initThread() {

        if("false".equals(isTest)){
            return;
        }
        YtcHandler test1 = new YtcHandler();
        Thread thread1 = new Thread(test1);
        thread1.setName("ytcHandlerThread1");
        thread1.start();

    }

    @Override
    public void run() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.21.17.17:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-test");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 90000);
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = "test";
        this.consumer.subscribe(Collections.singletonList("test"));
        while (true) {
            boolean success = true;
            ConsumerRecords<String, String> messages = consumer.poll(Duration.ofMillis(100L));
            List<BlockMileage> results = new ArrayList<>();
            for (ConsumerRecord<String, String> message : messages) {
                System.out.println(message.value());
                results.add(JSONObject.parseObject(message.value(), BlockMileage.class));
            }
            //写入数据库
            if (!CollectionUtils.isEmpty(results)) {
                try {
                   // blockMileageService.saveBatch(results);
                    consumer.commitSync();
                } catch (Exception e) {
                    log.error("blockMileage commit offset failed");
                    //success = false;
                }

            }
        }
    }

    @Override
    public void destroy() throws Exception {
        this.flag = false;
    }
    public static void main(String args[]) {

    }
}
